#include "RedisRegister.h"
#include "../rpcserver/rpcserver.h"
#include "tinyrpc/net/coroutine_function.h"
#include "tinyrpc/net/inetaddress.h"
#include "tinyrpc/rpc/rpcclient/rpcService.h"
#include "tinyrpc/base/stringUtil.h"
#include "tinyrpc/base/lexicalCast.hpp"
#include "tinyrpc/base/logger/logger.h"
namespace TinyRPC{
RedisRegister::RedisRegister()
:m_quit(false){

}
void RedisRegister::regist(std::unordered_map<std::string, ServiceInfo>& serviceMap, const INetAddress& serverAddress, int keyExpiredTime){
    std::string field = serverAddress.toString();
    for(auto& it : serviceMap){
        std::string key = KeyPrefix + it.first;
        
        m_redisClient.hset(key , field, "0");
        //m_redisClient.set(key + ":" + field,  "0");
        m_redisClient.setex(key + ":" + field, keyExpiredTime, "0");
        m_serviceList.insert({key, field});
        //m_subscribeChannels.push_back(key);
    }

}

void RedisRegister::delService(std::unordered_map<std::string, ServiceInfo>& serviceMap, const INetAddress& serverAddress){
    std::string field = serverAddress.toString();
    for(auto& it : serviceMap){
        std::string key = KeyPrefix + it.first;
        
        m_redisClient.hdel(key , field);
        //m_redisClient.set(key + ":" + field,  "0");
        m_redisClient.del(key + ":" + field);
        //m_subscribeChannels.push_back(key);
    }

}


void RedisRegister::co_updateServiceInfo(std::string serviceName, RPCServiceInfoPtr serviceMap, bool isInCo){
    if(isInCo){serviceMap->m_lock.lock();}
    //LockGuard<CoLock> gurad(serviceMap->m_lock);
    std::string key = KeyPrefix + serviceName;
    
    std::vector<string>serviceIPs = m_redisClient.hgetall(key);

    std::unordered_map<std::string, RPCServicePtr>& local_services = serviceMap->m_services;
    // for(auto& it : local_services){
    //     it.second->setStatus(RPCServiceStatus::DEAD);
    // }
    // 0.0.0.0
    for(auto& ip : serviceIPs ){
        if(ip.size()<7) continue;
        std::string result = m_redisClient.get(key + ":" + ip);
        if(result.size()>0){
            if(local_services.find(ip) == local_services.end()){
                LOG_INFO<<ip;
                vector<string> spiltStr;
                splitString(ip,  spiltStr, ":");
                INetAddress addr(spiltStr[0], LexicalCast<int>()(spiltStr[1]));
                RPCServicePtr rpcService = std::make_shared<RPCService>(addr);
                local_services[ip] = rpcService;
                LOG_INFO<<ip;
            }
        }

    }

    
    LOG_INFO<<"before clear dead";
    for(auto it=local_services.begin(); it != local_services.end(); ){
        if(it->second->getStatus() == RPCServiceStatus::DEAD){
            //it->second->closeConn();
            std::string result = m_redisClient.get(KeyPrefix + serviceName + ":" + it->first);
            if(result.size()==0){
                m_redisClient.hdel(KeyPrefix + serviceName, it->first);
                it = local_services.erase(it);
            }
            
        }else{
            it++;
        }
    }
    serviceMap->idx = serviceMap->m_services.begin();
    if(isInCo){serviceMap->m_lock.unlock();}
    
}


//心跳包， 这个需要在协程里面使用
void RedisRegister::co_heartbeat(int heartbeatTime, int keyExpiredTime){
    while (!m_quit)
    {
        for(auto& it : m_serviceList){
            // std::string key = it.first;
            // key += it->second;
            m_redisClient.expire(it.first + ":" + it.second, keyExpiredTime);
        }
        //LOG_INFO<<"heartbeat";
        coroutine_sleep(heartbeatTime);
    }
}


bool RedisRegister::publish(const std::string& channel, const std::string& message){
   return m_redisClient.publish(channel, message);
}

std::string RedisRegister::co_subscribe( SocketChannelPtr socketChannel,const std::vector<std::string>&channels, const Task& task){
   return  m_redisClient.co_subscribe(socketChannel, channels, task);
}

}